package task

import (
	"sync"
	"sync/atomic"
	"time"
)

type taskMonitor struct {
	AddedCount      int64         // 添加任务的累计次数
	TimeoutCount    int64         // 执行超时累计次数
	FailCount       int64         // 执行失败累计次数
	FinishCount     int64         // 执行完成累计次数
	AvgConsuming    time.Duration // 执行平均耗时
	MaxConsuming    time.Duration // 执行最长耗时
	MaxConsumingKey string        // 执行最长耗时的任务Key
	workerCount     int
	updateCh        chan *TaskInfo
	stopCh          chan struct{}
	isRuning        *atomic.Bool
}
type TaskMonitorData struct {
	AddedCount      int64         `json:"added_count,string"`   // 添加任务的累计次数
	TimeoutCount    int64         `json:"timeout_count,string"` // 执行超时累计次数
	FailCount       int64         `json:"fail_count,string"`    // 执行失败累计次数
	FinishCount     int64         `json:"finish_count,string"`  // 执行完成累计次数
	AvgConsuming    time.Duration `json:"avg_consuming,string"` // 执行平均耗时
	MaxConsuming    time.Duration `json:"max_consuming,string"` // 执行最长耗时
	MaxConsumingKey string        `json:"max_consuming_key"`    // 执行最长耗时的任务Key
	IsRuning        bool          `json:"is_runing"`
}

// 创建任务监控
//   - workerCount 任务并发执行协程数量(与TaskScheduled保持一致)
func newTaskMonitor(workerCount int) *taskMonitor {
	if workerCount == 0 {
		workerCount = 1024
	}
	tm := &taskMonitor{
		workerCount: workerCount,
		isRuning:    new(atomic.Bool),
	}
	tm.isRuning.Store(false)
	return tm
}

func (tm *taskMonitor) Run() {
	var wg sync.WaitGroup
	wg.Add(1)
	if !tm.isRuning.CompareAndSwap(false, true) {
		wg.Done()
		return
	}
	go func() {
		tm.updateCh = make(chan *TaskInfo, tm.workerCount)
		tm.stopCh = make(chan struct{})
		wg.Done()
		for {
			select {
			case t, ok := <-tm.updateCh:
				// 更新耗时, 使用缓冲通道处理并发安全
				if !ok {
					// 缓冲通道已关闭
					return
				}
				tm.updateConsuming(t)
			case <-tm.stopCh:
				// 收到stopCh 停止运行
				return
			}
		}
	}()
	wg.Wait()
}
func (tm *taskMonitor) Stop() {
	if !tm.isRuning.CompareAndSwap(true, false) {
		return
	}
	close(tm.stopCh)
	close(tm.updateCh)
}
func (tm *taskMonitor) IncAdded() {
	if tm.isRuning.Load() {
		atomic.AddInt64(&tm.AddedCount, 1)
	}
}
func (tm *taskMonitor) IncTimeout(t *TaskInfo) {
	// 更新耗时, 发送任务到 updateCh 缓冲通道（并发安全)
	if tm.isRuning.Load() {
		atomic.AddInt64(&tm.TimeoutCount, 1)
		tm.updateCh <- t
	}
}
func (tm *taskMonitor) IncFail(t *TaskInfo) {
	// 更新耗时, 发送任务到 updateCh 缓冲通道（并发安全)
	if tm.isRuning.Load() {
		atomic.AddInt64(&tm.FailCount, 1)
		tm.updateCh <- t
	}
}
func (tm *taskMonitor) IncFinish(t *TaskInfo) {
	// 更新耗时, 发送任务到 updateCh 缓冲通道（并发安全)
	if tm.isRuning.Load() {
		atomic.AddInt64(&tm.FinishCount, 1)
		tm.updateCh <- t
	}
}

func (tm *taskMonitor) updateConsuming(t *TaskInfo) {
	if t == nil || t.LastConsuming <= 0 || !tm.isRuning.Load() {
		return
	}
	// 更新平均耗时
	count := time.Duration(max(atomic.LoadInt64(&tm.FinishCount), atomic.LoadInt64(&tm.TimeoutCount), atomic.LoadInt64(&tm.FailCount)))
	tm.AvgConsuming = (tm.AvgConsuming*count + t.LastConsuming) / (count + 1)
	// 更新最长耗时
	if t.LastConsuming > tm.MaxConsuming {
		tm.MaxConsuming = t.LastConsuming
		tm.MaxConsumingKey = t.Key
	}
}
func (tm *taskMonitor) Data() TaskMonitorData {
	return TaskMonitorData{IsRuning: tm.isRuning.Load(),
		AddedCount:      atomic.LoadInt64(&tm.AddedCount),
		FailCount:       atomic.LoadInt64(&tm.FailCount),
		FinishCount:     atomic.LoadInt64(&tm.FinishCount),
		TimeoutCount:    atomic.LoadInt64(&tm.TimeoutCount),
		AvgConsuming:    tm.AvgConsuming,
		MaxConsuming:    tm.MaxConsuming,
		MaxConsumingKey: tm.MaxConsumingKey,
	}
}
